package com.fanxuankai.boot.canal.client;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.BooleanUtil;
import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import com.fanxuankai.boot.canal.util.RedisKey;
import com.fanxuankai.boot.redis.RedisUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Canal 工作者
 *
 * @author fanxuankai
 */
public class CanalWorker implements ApplicationRunner, DisposableBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(CanalWorker.class);
    @Resource
    private CanalClientProperties properties;
    @Resource
    private OtterClient otterClient;
    private volatile boolean running;
    private final ExecutorService executorService;

    public CanalWorker() {
        this.executorService = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(),
                ThreadFactoryBuilder.create()
                        .setDaemon(true).setNamePrefix("CanalWorkerThreadPool").build());
    }

    private void start() {
        if (running) {
            return;
        }
        executorService.execute(this::tryStart);
        this.running = true;
    }

    private void tryStart() {
        String key = RedisKey.withPrefix("canal:serviceCache", properties.getId() + ":CanalRunning");
        LOGGER.info("[" + properties.getId() + "] " + "ping...");
        String host = NetUtil.getLocalhostStr();
        do {
            if (BooleanUtil.isTrue(RedisUtils.ValueOps
                    .setIfAbsent(key, host, Duration.ofSeconds(properties.getPreemptive().getTimeout())))) {
                Runtime.getRuntime().addShutdownHook(new Thread(() -> RedisUtils.KeyOps.delete(key)));
                LOGGER.info("[" + properties.getId() + "] " + "pong...");
                executorService.execute(() -> {
                    do {
                        ThreadUtil.sleep(properties.getPreemptive().getKeep(), TimeUnit.SECONDS);
                        RedisUtils.ValueOps.set(key, host, properties.getPreemptive().getTimeout(), TimeUnit.SECONDS);
                    } while (running);
                });
                otterClient.start();
                break;
            }
            ThreadUtil.sleep(properties.getPreemptive().getPing(), TimeUnit.SECONDS);
        } while (true);
    }

    @Override
    public void run(ApplicationArguments args) {
        start();
    }

    @Override
    public void destroy() {
        otterClient.stop();
        this.running = false;
    }
}
